package com.shujia.core

import org.apache.flink.streaming.api.scala._

object Demo4Parallelism {
  def main(args: Array[String]): Unit = {
    /**
      * flink 并行度
      * 1、再代码中设env.setParallelism(1),
      * 2、再提交任务的时候设计  -p  优先级最低
      * 3、每一个算子可以单独设置并行度  优先级最高
      *
      * 不建议再代码中设并行度，不方便优化
      *
      *
      * flink 启动任务需要多少资源有并行度决定，一个并行度需要一个slot
      *
      */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    linesDS
      .flatMap(_.split("."))
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)//.setParallelism(5)
      .print()

    env.execute("Demo4Parallelism")
  }
}
